package com.aispeech.asr.apis.detection.service;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.stereotype.Service;

import com.aispeech.asr.apis.detection.config.AppConfig;
import com.aispeech.asr.apis.detection.domain.EventDetectionContext;
import com.aispeech.asr.apis.detection.exceptions.BusinessException;
import com.aispeech.asr.apis.detection.form.DetectionForm;
import com.aispeech.asr.apis.detection.function.EventDetectionFunction;
import com.aispeech.asr.apis.detection.metrics.Prometheus;
import com.aispeech.asr.apis.detection.vo.AsrDataVo;
import com.aispeech.asr.comm.data.domain.meter.MeterType;
import com.aispeech.asr.comm.data.vo.Result;
import com.aispeech.asr.comm.enums.ErrorCode;
import com.aispeech.asr.webase.component.BigFileHttpPoster;
import com.aispeech.asr.webase.component.MeterSyncer;

import io.prometheus.client.Gauge;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

@Service
@Slf4j
public class AsrService {
    //    @Autowired
//    ExecutorService executorService;
    @Autowired
    BigFileHttpPoster bigFileHttpPoster;

    @Autowired
    ApplicationContext applicationContext;

    @Autowired
    AppConfig config;

    @Autowired
    MeterSyncer meterSyncer;

    Gauge.Child gaugeChildConcurrency;

    public Mono<Result> process(String sessionId, DetectionForm form, InputStream inputStream) throws IOException {
        long received_t = System.currentTimeMillis();

        EventDetectionContext asrContext = EventDetectionContext.of(
                sessionId,
                form,
                inputStream
        );
        
        
        String audioUrl = String.format("/audios/%s/%s.%s",
        		LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd")),
        		sessionId,
        		form.getAudio().getAudioType());
        
        gaugeChildConcurrency = Prometheus.GAUGE_N_CONCURRENCY_SESSIONS.labels(Prometheus.Env(), Prometheus.NodeName());
        gaugeChildConcurrency.inc();

        EventDetectionFunction detector = applicationContext.getBean(EventDetectionFunction.class);

        Scheduler scheduler = Schedulers.boundedElastic();

        return Mono.just(asrContext)                      // 1.初始化context，整个任务流程中随时可用
                .publishOn(scheduler)                     // 2.在独立线程中做音频切割的逻辑，避免锁netty线程
                .map(detector)                            // 3.音频解码和VAD切割
                //.map(cleaner)                           // 7.清理工作目录并备份
                .doOnError(throwable -> {
                	log.error("[ doOnError ]productId: {}, sessionId: {}",form.getProductId(), sessionId);
                    int errid = ErrorCode.ERR_UNKNOWN.errid;
                    AsrDataVo asrData = new AsrDataVo(asrContext.getSessionId(), asrContext.getForm());

                    if (throwable instanceof BusinessException) {
                    	BusinessException exception = (BusinessException) throwable;
                        errid = exception.getCode();
                        exception.setData(asrData);
                    }

                    asrContext.addMetric("finished");
                    addPrometheusMetrics(asrContext, errid, 0L);
                    asrData.setMetrics(asrContext.getMetrics());
//                    BigDataLogger.logSentence(
//                            errid,
//                            throwable.getMessage(),
//                            asrData
//                    );

                    if (!form.isDebug())
                        asrData.setMetrics(null);
                })
                .map(resultContext -> {
                	
                	AsrDataVo asrData = new AsrDataVo(resultContext.getSessionId(), resultContext.getForm());
                	
                	//asrData.setResult(resultContext.getResult().getData());
                	asrData.setResult(resultContext.getResultList());
                	
                    Result result = Result.ok(asrData);

                    asrContext.addMetric("finished");
                    addPrometheusMetrics(resultContext, 0, received_t);
                    asrData.setMetrics(resultContext.getMetrics());
                    //BigDataLogger.logSentence(result, asrContext.getRes()); //TODO

                    log.info("[ Finish ]productId: {}, sessionId: {}, rows: {}",
                            form.getProductId(), resultContext.getSessionId(), asrData.getResult().size());

                    if (!form.isDebug())
                        asrData.setMetrics(null);

                    return result;            // 8.返回最终结果
                }).doOnTerminate(() -> {
                    //backup(asrContext.getSessionId(), form.getAudio().getAudioType().toLowerCase(), inputStream);
                	try {
						log.info("[doOnTerminate] backup:isSaveAudio:{},size:{}",config.isSaveAudio(), inputStream.available());
					} catch (IOException e1) {
						// TODO Auto-generated catch block
						e1.printStackTrace();
					}
                	if (config.isSaveAudio())
                		
                        backup(audioUrl, inputStream);
                    try {
                        inputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                	
                    if (gaugeChildConcurrency != null && gaugeChildConcurrency.get() > 0)
                        gaugeChildConcurrency.dec();
                });
    }
    
    private void backup(String filePath, InputStream inputStream) {

        try {
            byte[] buff = new byte[inputStream.available()];
            inputStream.read(buff);
            bigFileHttpPoster.add(filePath, filePath, new ByteArrayResource(buff));
        } catch (IOException e) {
            log.warn("backup file: {} failed.", filePath);
        }
    }

    private void addPrometheusMetrics(EventDetectionContext asrContext, int errid, long received_t) {
        String productId = asrContext.getForm().getProductId();

        Prometheus.COUNTER_DETECTION_REQUESTS
                .labels(
                        Prometheus.Env(),
                        Prometheus.NodeName(),
                        productId,
                        String.valueOf(errid)
                ).inc();

        if ("278587440".equals(productId)) return; // 拨测的记录不计入实时率和计量。

        if (errid == 0) {
            long duration = asrContext.getForm().getDuration();
            long finished = System.currentTimeMillis();

            float elpsed = -1F;
            if (finished != 0 && received_t != 0 && finished > received_t) {
                elpsed = finished - received_t;
            }

            if (duration != 0 && elpsed > 0) {
                Prometheus.SUMMARY_REAILRATE.labels(
                        Prometheus.Env(),
                        Prometheus.NodeName()
                ).observe(elpsed / duration);
            }

            meterSyncer.reportMeter(MeterType.ASR_SED, productId, 1 , duration);
        }
    }
}
